use std::collections::hash_map::Entry;
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};

use serde::Deserialize;
use wtask_app::{AppTaskArchv, cli_app, msg_func_none};
use wtask_base::app::get_wtask_data_dir;
use wtask_base::encrypt::decrypt_str_to_str;
use wtask_base::error::{WError, WResult};
use wtask_base::export::{json, log_debug, log_error, serde_from_str};
use wtask_base::timer::{TimerAppTrait, TimerRun};
use wtask_file::{IncomingBody, Request, Response, StatusCode, header};

use crate::basic::{parse_body, response_body, MachineConfigs, ResponseBody};
use crate::handle::flow::convert::convert_to_wtask;
use crate::{basic::WEB_SESSION, handle::flow::{basic::{FlowData, FlowSubmitData}, storage::FlowStorage, timer::FlowApp}};







fn flow_check(flow: &FlowData, groups_id: &HashSet<String>, machines: &HashSet<String>) -> WResult<()> {
    let yes = flow.setting.groups.iter().any(| v | groups_id.contains(v));
    if !yes {
        log_debug!("{:?} vs {:?}", groups_id, flow.setting.groups);
        return Err(WError::DataError("Auth error: group".to_owned()));
    }
    let yes = flow.setting.machines.iter().all(| v | machines.contains(v));
    if !yes {
        return Err(WError::DataError("Auth error: machine".to_owned()));
    }
    Ok(())
}

async fn flow_log(run_id: &str, groups_id: &HashSet<String>, machines: &HashSet<String>) -> WResult<String> {
    let flow_storage = WEB_SESSION.flow.read().await;
    let flow_log = flow_storage.run.data
        .get(run_id)
        .ok_or(WError::DataError("log get error".to_owned()))?;
    flow_check(&flow_log.data, groups_id, machines)?;
    Ok(json!({
        "msg": "编排任务记录读取成功",
        "data": flow_log
    }).to_string())
}

async fn flow_read(flow_id: &str, groups_id: &HashSet<String>, machines: &HashSet<String>) -> WResult<String> {
    let flow_storage = WEB_SESSION.flow.read().await;
    let flow = flow_storage.data_get(flow_id)?;
    flow_check(flow, groups_id, machines)?;
    Ok(json!({
        "msg": "编排任务读取成功",
        "data": flow
    }).to_string())
}

async fn flow_list_meta(groups_id: &HashSet<String>, machines: &HashSet<String>) -> WResult<String> {
    let flow_storage = WEB_SESSION.flow.read().await;
    let flow = groups_id
        .iter()
        .flat_map(| v | flow_storage.group_get(v))
        .collect::<BTreeSet<&str>>()
        .iter()
        .map(| &v | flow_storage.data_get(v))
        .collect::<WResult<Vec<_>>>()?
        .into_iter()
        .filter(| &v | v.setting.machines.iter().all(| vv | machines.contains(vv)))
        .map(| v | {
            let log_temp = flow_storage.log_list(&v.setting.id);
            let running = log_temp.iter().any(| &vv | vv.running());
            let timer_res = flow_storage.timer.has.get(&v.setting.id).and_then(| timer_list | {
                let t = timer_list
                    .iter()
                    .filter_map(| v | flow_storage
                        .timer
                        .data
                        .get(v)
                        .map(| i | (
                            v,
                            json!({
                                "log": &i.1.len(),
                                "timer": TimerRun::<FlowApp>::parse_timer_basic(v).convert_to_display()
                            })
                        )
                    ))
                    .collect::<BTreeMap<_, _>>();
                if t.is_empty() {
                    None
                } else {
                    Some(t)
                }
            });
            json!({ "flow": v, "running": running, "log": log_temp, "timer": timer_res })
        })
        .collect::<Vec<_>>();
    Ok(json!({
        "msg": "编排任务列表读取成功",
        "data": flow
    }).to_string())
}

async fn flow_list_simple(groups_id: &HashSet<String>, machines: &HashSet<String>) -> WResult<String> {
    let flow_storage = WEB_SESSION.flow.read().await;
    let flow = groups_id
        .iter()
        .flat_map(| v | flow_storage.group_get(v))
        .collect::<BTreeSet<&str>>()
        .iter()
        .map(| &v | flow_storage.data_get(v))
        .collect::<WResult<Vec<_>>>()?
        .into_iter()
        .filter(| &v | v.setting.machines.iter().all(| vv | machines.contains(vv)))
        .collect::<Vec<_>>();
    Ok(json!({
        "msg": "编排任务列表读取成功",
        "data": flow
    }).to_string())
}


async fn flow_save(flow: FlowData, groups_id: &HashSet<String>, machines: &HashSet<String>) -> WResult<String> {
    let r = {
        let flow_storage = WEB_SESSION.flow.read().await;
        let is_old = flow_storage.storage.contains_key(&flow.setting.id);
        if is_old {
            // 旧的，需要检查权限
            let flow_old = flow_storage.data_get(&flow.setting.id)?;
            flow_check(flow_old, groups_id, machines)?;
        }
        json!({
            "msg": format!("编排任务“{}”保存成功", flow.setting.name),
            "data": flow
        }).to_string()
    };
    // 数据
    let mut flow_storage = WEB_SESSION.flow.write().await;
    let id = flow.setting.id.clone();
    let groups = flow.setting.groups.clone();
    flow_storage.storage.insert(id.clone(), flow);
    for g in groups {
        match flow_storage.groups.entry(g) {
            Entry::Occupied(mut v) => v.get_mut().insert(id.clone()),
            Entry::Vacant(v) => v.insert(HashSet::new()).insert(id.clone()),
        };
    }
    Ok(r)
}


async fn flow_status(run_id: &str, groups_id: &HashSet<String>, machines: &HashSet<String>) -> WResult<String> {
    let s = WEB_SESSION.flow.read().await;
    // status
    let (flow_status, flow_data) = s.run.data
            .get(run_id)
            .map(| v | (&v.status, &v.data))
            .ok_or(WError::DataError("status get error".to_owned()))?;
    // check auth
    flow_check(flow_data, groups_id, machines)?;
    // res
    let r = json!({
        "msg": "状态获取成功",
        "data": flow_status
    }).to_string();
    Ok(r)
}

#[allow(clippy::too_many_arguments)]
async fn flow_run(flow_id: &str, groups_id: &HashSet<String>, machines: &HashSet<String>, flow_submit_data: FlowSubmitData, token: &str, timer: Option<String>) -> WResult<String> {
    // check auth
    let flow = WEB_SESSION.flow.read().await.data_get(flow_id)?.clone();
    flow_check(&flow, groups_id, machines)?;
    
    let flow_name = flow.setting.name.clone();

    // machines
    let machines = {
        let login_storage = WEB_SESSION.login.read().await;
        let machines_raw = login_storage.as_ref().get(token).unwrap();
        flow.setting.machines.clone()
            .into_iter()
            .map(| v | {
                let c = machines_raw.get(&v).unwrap().config.clone();
                (v, c)
            })
            .collect::<MachineConfigs>()
    };

    // app
    let app = FlowApp {
        data: flow,
        submit: flow_submit_data,
        machines,
    };
    
    // run
    let r = if let Some(timer) = timer {
        log_debug!("submit timer: {:?}", timer);
        let app_timer = TimerRun::new(&timer, app)?;
        let timer_id = app_timer.name.clone();
        let r = json!({
            "msg": format!("定时任务{:?}提交成功", flow_name),
            "data": timer_id
        }).to_string();
        log_debug!("submit timer check: {:?}", timer_id);
        let app_timer_config = app_timer.extract_config();
        WEB_SESSION.flow.write().await.timer.insert(timer_id, (app_timer_config.app, app_timer_config.log));
        r
    } else {
        let run_id = app.run().await?;
        json!({
            "msg": format!("{:?} 开始运行", flow_name),
            "data": run_id
        }).to_string()
    };

    Ok(r)
}



async fn flow_log_del_inner(flow_storage: &mut FlowStorage, run_id: &str, token: &str, del: bool) {
    log_debug!("log del inner: {} {} ...", run_id, del);
    let run_log = flow_storage.run.remove(run_id);
    if run_log.is_none() {
        return ;
    }
    let mut run_log = run_log.unwrap();
    // machine config from login
    let machine_configs = {
        let login_storage = WEB_SESSION.login.read().await;
        let machines_raw = login_storage.as_ref().get(token).unwrap();
        run_log.data.setting.machines
            .into_iter()
            .map(| v | {
                let c = machines_raw.get(&v).unwrap().config.clone();
                (v, c)
            })
            .collect::<MachineConfigs>()
    };
    // iter task_id for machine
    for (node_id, task_data) in run_log.status.task.drain() {
        for machine_id in task_data.dir_result.keys() {
            let machine_config = machine_configs.get(machine_id);
            if let Some(machine_config) = machine_config {
                log_debug!("log del action: {} {} {} ...", run_id, task_data.task_id, del);
                let _ = cli_app(AppTaskArchv {
                    del,
                    id: task_data.task_id.clone()
                }, machine_config, None, msg_func_none).await;
            } else {
                log_error!("machine config not found: {} {} {}", run_id, node_id, task_data.task_id);
            }
        }
    }
}


async fn flow_timer_del_inner(flow_storage: &mut FlowStorage, timer_id: &str, token: &str, del_log: bool, archv_log: bool) {
    if let Some(timer) = flow_storage.timer.remove(timer_id)
        && del_log {
            for run_id in timer.1 {
                flow_log_del_inner(flow_storage, &run_id, token, !archv_log).await;
            }
        }
}


async fn flow_log_del(run_id: &str, groups_id: &HashSet<String>, machines: &HashSet<String>, token: &str, del: bool) -> WResult<String> {
    // check auth
    {
        let flow_storage = WEB_SESSION.flow.read().await;
        let run_log = flow_storage.run.data.get(run_id);
        if run_log.is_none() {
            return Ok(json!({
                "msg": "运行记录不存在",
                "data": "{}"
            }).to_string());
        }
        let run_log = run_log.unwrap();
        flow_check(&run_log.data, groups_id, machines)?;
    }
    let mut flow_storage = WEB_SESSION.flow.write().await;
    flow_log_del_inner(&mut flow_storage, run_id, token, del).await;
    Ok(json!({
        "msg": "删除成功",
        "data": run_id,
    }).to_string())
}



async fn flow_timer_del(timer_id: &str, groups_id: &HashSet<String>, machines: &HashSet<String>, token: &str, del_log: bool, archv_log: bool) -> WResult<String> {
    // check auth
    {
        let flow_storage = WEB_SESSION.flow.read().await;
        let timer = flow_storage.timer.data.get(timer_id);
        if timer.is_none() {
            return Ok(json!({
                "msg": "定时任务不存在",
                "data": "{}"
            }).to_string());
        }
        let timer = timer.unwrap();
        flow_check(&timer.0.data, groups_id, machines)?;
    }
    
    let mut flow_storage = WEB_SESSION.flow.write().await;
    flow_timer_del_inner(&mut flow_storage, timer_id, token, del_log, archv_log).await;
    Ok(json!({
        "msg": "删除成功",
        "data": timer_id,
    }).to_string())
}


async fn flow_del_inner(flow_id: &str, del_log: bool, archv_log: bool, token: &str) -> WResult<Option<FlowData>> {
    let mut flow_storage = WEB_SESSION.flow.write().await;
    // 删除
    // data
    let r = if let Some(flow_data) = flow_storage.storage.remove(flow_id) {
        // groups
        for group_temp in flow_data.setting.groups.iter() {
            if let Some(gs) = flow_storage.groups.get_mut(group_temp) {
                let _ = gs.remove(flow_id);
            }
        }
        // timer
        if let Some(timer_ids) = flow_storage.timer.has.get(flow_id).cloned() {
            for timer_id_temp in timer_ids {
                flow_timer_del_inner(&mut flow_storage, &timer_id_temp, token, del_log, archv_log).await;
            }
        }
        // other run
        if let Some(run_ids) = flow_storage.run.has.get(flow_id).cloned() {
            for run_id_temp in run_ids {
                if del_log {
                    // del run-log
                    flow_log_del_inner(&mut flow_storage, &run_id_temp, token, !archv_log).await;
                }
            }
        }
        Some(flow_data)
    } else {
        None
    };
    Ok(r)
}


async fn flow_del_check(flow_id: &str, groups_id: &HashSet<String>, machines: &HashSet<String>) -> WResult<FlowData> {
    let flow_storage = WEB_SESSION.flow.read().await;
    let flow = flow_storage.data_get(flow_id)?;
    flow_check(flow, groups_id, machines)?;
    let is_running = flow_storage.log_list(flow_id).into_iter().any(| v | v.running());
    if is_running {
        return Err(WError::DataError(format!("编排任务“{}”正在运行，无法删除", flow.setting.name)))
    }
    Ok(flow.clone())
}


async fn flow_del(flow_id: &str, groups_id: &HashSet<String>, machines: &HashSet<String>, token: &str, del_log: bool, archv_log: bool) -> WResult<String> {
    // check auth
    flow_del_check(flow_id, groups_id, machines).await?;
    
    let r = if let Some(flow_data) = flow_del_inner(flow_id, del_log, archv_log, token).await? {
        json!({
            "msg": format!("编排任务“{}”删除成功", flow_data.setting.name),
            "data": "{}"
        })
    } else {
        json!({
            "msg": "编排任务不存在",
            "data": "{}"
        })
    };
    Ok(r.to_string())
}


#[derive(Debug, Deserialize)]
struct FlowConvertData {
    // 保存结果任务的machine id
    machine_save: String,
    // 是否删除该flow
    del: bool,
    // 删除该flow时是否删除运行记录
    del_log: bool,
    // 删除该flow时是否存档运行记录
    archv_log: bool,
}


/// 将flow转换为wtask
/// 
/// 1. download all task
/// 2. write meta files:
///     1. readme: flow.note
///     2. meta: name: flow.name, tags: flow.groups.name, args: flow.args
///     3. run.sh: flatten wtask / shell
///     4. install.sh: flatten wtask 
async fn flow_convert_task(flow_id: &str, groups_id: &HashSet<String>, machines: &HashSet<String>, token: &str, data_convert: FlowConvertData) -> WResult<String> {
    // check auth
    let flow_data = flow_del_check(flow_id, groups_id, machines).await?;
    
    // machine configs
    let machine_configs = {
        let s = WEB_SESSION.login.read().await;
        let machines = s.as_ref()
            .get(token)
            .ok_or(WError::DataError("No login".to_owned()))?;
        flow_data.setting.machines
            .iter()
            .filter_map(| v | machines.get(v).map(| vv | (v.clone(), vv.config.clone())))
            .collect::<HashMap<_, _>>()
    };

    let dir_task = get_wtask_data_dir("temp", true)?.join(&flow_data.setting.id);

    let res = match convert_to_wtask(&dir_task, flow_data, machine_configs, data_convert.machine_save).await {
        Ok(task_meta) => {
            if data_convert.del {
                flow_del_inner(
                    flow_id,
                    data_convert.del_log,
                    data_convert.archv_log,
                    token
                ).await?;
            }
            json!({
                "msg": "编排任务转换成功",
                "data": {
                    "status": true,
                    "data": task_meta,
                }
            })
        },
        Err(e) => {
            json!({
                "msg": "编排任务转换失败",
                "data": {
                    "status": false,
                    "data": e.to_string(),
                }
            })
        }
    };
    // let _ = remove_dir_all(dir_task);

    Ok(res.to_string())
}




pub(crate) async fn response_flow(req: Request<IncomingBody>, token: &str) -> WResult<Response<ResponseBody>> {
    let flow_action = req.headers().get("Action").cloned().ok_or(WError::Empty)?;
    
    let flow_id = req.headers().get("Flow").cloned();
    let flow_timer = if let Some(timer) = req.headers().get("Timer") {
        timer.to_str().ok().map(|t| t.to_owned())
    } else {
        None
    };

    let groups_id = {
        let groups_id = req.headers().get("Groups").cloned().ok_or(WError::Empty)?;
        let groups_id = decrypt_str_to_str(groups_id.to_str().unwrap(), token)?;
        serde_from_str::<HashSet<String>>(&groups_id)?
    };

    // login
    let machines = {
        let s = WEB_SESSION.login.read().await;
        let machines = s.as_ref().get(token);
        if machines.is_none() {
            let json = json!({
                "msg": "登录失败",
                "data": "No machine found",
            }).to_string();
            let res = Response::builder()
                .status(StatusCode::UNAUTHORIZED)
                .header(header::CONTENT_TYPE, "application/json")
                .body(response_body(&json, token)).unwrap();
            return Ok(res);
        }
        machines.unwrap()
            .keys()
            .cloned()
            .collect::<HashSet<String>>()
    };
    
    let res = match (flow_action.to_str().unwrap(), flow_id) {
        ("status", Some(v)) => flow_status(v.to_str().unwrap(), &groups_id, &machines).await,
        ("list-simple", _) => flow_list_simple(&groups_id, &machines).await,
        ("list-meta", _) => flow_list_meta(&groups_id, &machines).await,
        ("read", Some(v)) => flow_read(v.to_str().unwrap(), &groups_id, &machines).await,
        ("save", _) => {
            let flow_temp = parse_body(req, token).await?;
            flow_save(flow_temp, &groups_id, &machines).await
        },
        ("run", Some(v)) => {
            let flow_submit_data = parse_body(req, token).await?;
            flow_run(v.to_str().unwrap(), &groups_id, &machines, flow_submit_data, token, flow_timer).await
        },
        ("log", Some(v)) => flow_log(v.to_str().unwrap(), &groups_id, &machines).await,
        ("log-del", Some(v)) => flow_log_del(v.to_str().unwrap(), &groups_id, &machines, token, true).await,
        ("log-archv", Some(v)) => flow_log_del(v.to_str().unwrap(), &groups_id, &machines, token, false).await,
        ("timer-remove", Some(v)) => flow_timer_del(v.to_str().unwrap(), &groups_id, &machines, token, false, true).await,
        ("timer-del", Some(v)) => flow_timer_del(v.to_str().unwrap(), &groups_id, &machines, token, true, false).await,
        ("timer-archv", Some(v)) => flow_timer_del(v.to_str().unwrap(), &groups_id, &machines, token, true, true).await,
        ("flow-remove", Some(v)) => flow_del(v.to_str().unwrap(), &groups_id, &machines, token, false, true).await,
        ("flow-del", Some(v)) => flow_del(v.to_str().unwrap(), &groups_id, &machines, token, true, false).await,
        ("flow-archv", Some(v)) => flow_del(v.to_str().unwrap(), &groups_id, &machines, token, true, true).await,
        ("flow-convert", Some(v)) => {
            let data_convert = parse_body(req, token).await?;
            flow_convert_task(v.to_str().unwrap(), &groups_id, &machines, token, data_convert).await
        },
        _ => {
            let json = json!({
                "msg": "数据错误",
                "data": "No action match or No flow_id found;",
            }).to_string();
            let res = Response::builder()
                .status(StatusCode::UNAUTHORIZED)
                .header(header::CONTENT_TYPE, "application/json")
                .body(response_body(&json, token)).unwrap();
            return Ok(res);
        }
    };

    let (status, json) = match res {
        Ok(v) => (StatusCode::OK, v),
        Err(WError::DataError(v)) => (StatusCode::BAD_REQUEST, json!({
            "msg": "操作错误",
            "data": v,
        }).to_string()),
        _ => (StatusCode::BAD_REQUEST, json!({
            "msg": "未知错误",
            "data": "",
        }).to_string()),
    };
    let response = Response::builder()
        .status(status)
        .header(header::CONTENT_TYPE, "application/json")
        .body(response_body(&json, token)).unwrap();
    Ok(response)
}

